//
//  IvanNetStream.m
//  YuchBox
//
//  Created by Ivan on 11-10-5.
//  Copyright 2011年 __MyCompanyName__. All rights reserved.
//



#import "IvanNetStream.h"
#import "SingletonClassGenerator.h"
#import "StreamPeer.h"

@implementation IvanNetStream
@synthesize packetHeaderSize = _packetHeaderSize;

SYNTHESIZE_SINGLETON_FOR_CLASS(IvanNetStream);
- (void)initMembers
{
    _packetHeaderSize   = PACKET_HEAD_SIZE;
    //_isFirstFourBytes = YES;
    _wbHead             = 0;
    _wbTail             = 0;
    _hasEstablished     = NO;
}

- (id)init
{
    self = [super init];
    if (self) {
        // Initialization code here.
        [self initMembers];
    }
    
    return self;
}

- (id)initSingleton
{
    self = [self init];
    return  self;
}

- (void) dealloc
{
//    NSLog(@"dealloc %d", [IVAN_NET_STREAM retainCount]);
//    NSLog(@"////////////////////!!!!!!!!!!!!!!!!");
    [self closeStreams];
    [super dealloc];
    
    // WARNING the singleton class must be set nil!!!!
    sharedIvanNetStream = nil;
}

- (Boolean)startConnect
{
        //NSLog(@"%u", [IVAN_NET_STREAM retainCount]);
    CFReadStreamRef     readStream = NULL;
    CFWriteStreamRef    writeStream = NULL;
    
    // WARNING just create the stream, not real connect to the net
    CFStreamCreatePairWithSocketToHost(kCFAllocatorDefault,
                                       (CFStringRef)_serverURL,
                                       _port,
                                       &readStream,
                                       &writeStream);
    if (readStream && writeStream) {
       // TTLOG_INFO(@"Connect successful!");
        _inStream  = (NSInputStream *)readStream;
        _outStream = (NSOutputStream *)writeStream;
        [_inStream setProperty:NSStreamNetworkServiceTypeVoIP forKey:NSStreamNetworkServiceType];
        [_outStream setProperty:NSStreamNetworkServiceTypeVoIP forKey:NSStreamNetworkServiceType];
    } else {
        //Error Control
        return FALSE;
    }
    
    return TRUE;
}

- (Boolean)openStreams
{
    [_inStream  retain];
    [_outStream retain];
    
    // For SSL
    //[_inStream  setProperty:NSStreamSocketSecurityLevelSSLv3 forKey:NSStreamSocketSecurityLevelKey];
    //[_outStream setProperty:NSStreamSocketSecurityLevelSSLv3 forKey:NSStreamSocketSecurityLevelKey];
    // CFWriteStreamSetProperty((CFWriteStreamRef)_outStream, kCFStreamPropertySSLSettings, [NSMutableDictionary dictionaryWithObjectsAndKeys:(id)kCFBooleanFalse,kCFStreamSSLValidatesCertificateChain,kCFBooleanFalse,kCFStreamSSLIsServer,nil]);
    [_inStream  setDelegate:self];
    [_inStream  scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
    [_inStream  open];
    [_outStream setDelegate:self];
    //[_outStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
    [_outStream open];
    
    TTLOG_INFO(@"Open stream successful!");
    return TRUE;
}

- (void)closeStreams
{
    // PromptView sharedStream
    // [[PromptView sharedPromptView]dismissPromptView];
    _run = NO;
    
//    [_asyncListener release];
    [_inStream  close];
    [_outStream close];
    [_inStream  removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
   // [_outStream removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
    [_inStream  setDelegate:nil];
    [_outStream setDelegate:nil];
    [_inStream  release];
    [_outStream release];
    _inStream  = nil;
    _outStream = nil;
    
    [_dataBuffer release];
}


- (void)asyncListener
{
    static int restart_time = 1;
    [[NSThread currentThread] setName:[NSString stringWithFormat:@"async%d", restart_time++]];
    
    TTLOG_INFO(@"AsyncListener start...");
    _runLoopOfAsyncListener = [NSRunLoop currentRunLoop];
    [_inStream  setDelegate:self];
    [_inStream  scheduleInRunLoop:_runLoopOfAsyncListener forMode:NSDefaultRunLoopMode];
    [_inStream  open];
    
    [_outStream setDelegate:self];
    //[_outStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
    [_outStream open];
    
    NSMachPort *dummyPort = [[NSMachPort alloc] init];	

    [_runLoopOfAsyncListener addPort:dummyPort forMode:NSDefaultRunLoopMode];
    [dummyPort release];

    _run = YES;
    while (_run){		
		NSAutoreleasePool *loopPool = [[NSAutoreleasePool alloc] init];		
		[_runLoopOfAsyncListener runMode:NSDefaultRunLoopMode beforeDate:[NSDate distantFuture]];
		[loopPool drain];
	} 

    TTLOG_INFO(@"AsyncListener finish!!");
}

- (Boolean)connectWithServer:(NSString *)server port:(CFIndex)port delegate:(NSObject <IvanNetStreamDelegate> *)delegate 
{
    if (![self isServerAvailable]) {
        // TODO Error Control 
        return FALSE;
        //[self reconnect];
    }
    
    if (nil == _inStream || nil == _outStream) {
        _serverURL       = server;
        _port            = port;
        _currentDelegate = delegate;
    
        TTLOGF_INFO(@"start connect to server(%@:%d)...", server, port);
        if (![IVAN_NET_STREAM startConnect]) {
            NSInteger reconnectCounter = 0;
            
            // not real reconnect here 
            while ((++reconnectCounter) < MAX_RECONNECT_TIMES) {
                if ([IVAN_NET_STREAM startConnect]) {
                    break;
                }
                sleep(reconnectCounter);
            }
            
            if (reconnectCounter == MAX_RECONNECT_TIMES) {
                NSLog(@"Try my best!");
                return FALSE;
            }
        }
        
        //  start the listener to read from and write to the network
        [self performSelectorInBackground:@selector(asyncListener) withObject:nil];
    }    
    
    return TRUE;
}

- (Boolean)reconnect
{
    return [self connectWithServer:_serverURL port:_port delegate:_currentDelegate];
}

- (void)disconnect
{
    if (_wbHead != _wbTail) {
        TTLOG_WARN(@"Remain data havn't sent yet!!");
        [self initMembers];
    }
    [self closeStreams];
}

- (BOOL)isServerAvailable
{
    // TODO check the Device the network is availiable
    return TRUE;
    NSString *addressString = _serverURL;
    
    if(!addressString)
    {
        return NO;
    }
    
    /* 
     SCNetworkReachabilityRef defaultRouteReachability = SCNetworkReachabilityCreateWithName(kCFAllocatorDefault, [addressString UTF8String]);
     SCNetworkReachabilityFlags flags;
     
     
     BOOL didRetrieveFlags = SCNetworkReachabilityGetFlags( defaultRouteReachability,  &flags);
     CFRelease(defaultRouteReachability);   
     if(!didRetrieveFlags)
     {
     return NO;
     }
     
     
     BOOL isReachable = flags & kSCNetworkFlagsReachable;  
     BOOL needsConnection =flags & kSCNetworkFlagsConnectionRequired;
     
     return (isReachable && !needsConnection) ?  YES : NO;
     */
    return NO;
}

- (void)stream:(NSStream*)aStream handleEvent:(NSStreamEvent)eventCode 
{
    switch(eventCode) {                   
        case  NSStreamEventOpenCompleted: {         // connect and stream is open completed 
            TTLOG_DEBUG(@"NSStreamEventOpenCompleted");
            assert(aStream == _outStream || aStream == _inStream);
            _hasEstablished = YES;
            break;  
        }  
        case NSStreamEventHasSpaceAvailable: {      // stream is ready for write
            TTLOG_DEBUG(@"NSStreamEventHasSpaceAvailable, write stream is ready");
            if (aStream != _outStream) {
                [self dumpToBuffer:(NSOutputStream *)aStream];
            }
        
//            TTLOGF_INFO(@"Buffer remain <_wbHead:%d, _wbTail:%d, datasize:%d, remain:%d", _wbHead, _wbTail, _wbTail - _wbHead, BUFFER_SIZE - _wbTail);
            if (_wbHead == _wbTail) {
                if (aStream != _outStream) {
                    [aStream close];
                }
                [aStream removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
                return;
            }
            
            // wait for the condition. write lock
#define MAX_READ_TIMES  5
            int readTimes = 0;
            while (![_outStream hasSpaceAvailable]){
                TTLOG_WARN(@"Outstream not ready");
                if (readTimes++ < MAX_READ_TIMES) {
                    sleep(readTimes);
                } else {
                    break;
                }
            }
            
            NSInteger dataLen       = *((NSInteger *)&_writeBuffer[_wbHead]);
            NSInteger packetLen     = dataLen + _packetHeaderSize;
            
//            TTDUMP(&_writeBuffer[_wbHead], MIN(20, packetLen));
            NSInteger outLen        = 0;
            NSData *outData         = [_currentDelegate preProcessData:&_writeBuffer[_wbHead] length:packetLen outLength:&outLen];

            [StreamPeer writeBytesWithOutStream:_outStream ptr:(UInt8 *)[outData bytes] maxLength:outLen];
            
            _wbHead += packetLen;
            
            _wbTail %= BUFFER_SIZE;
            _wbHead %= BUFFER_SIZE;
            
            if (_wbHead <= _wbTail) {
                // send OK
//                TTLOGF_INFO(@"Buffer remain <_wbHead:%d, _wbTail:%d, datasize:%d, remain:%d", _wbHead, _wbTail, _wbTail - _wbHead, BUFFER_SIZE - _wbTail);
            } else { 
                TTLOG_ERROR(@"<_wbHead > _wbTail>, something error occur!");
            } 
            
            if (aStream != _outStream) {
                [aStream close];
            }
            [aStream removeFromRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
            break;   
        }    
        case NSStreamEventHasBytesAvailable: {      // packge arrival from server
            TTLOG_DEBUG(@"NSStreamEventHasBytesAvailable, data arrival");
            // ~WARNING if not sleep, we will get 3 extra bytes in the stream. Very strange!!!
            // Fixed by TZZ. server problem
            [_currentDelegate readInStream:_inStream];
            break;
        }     
        case NSStreamEventEndEncountered:{          // disconnect   
             NSError *error = [aStream streamError];
            if ([error code]) {
                TTLOGF_WARN(@"NSStreamEventEndEncountered, the server disconnect <%d:%@>", [[aStream streamError] code], [error localizedDescription]);
//                [self closeStreams];
                [self disconnect];
                [_currentDelegate callBackHandle:IvanNSEventErrorServerRefuse info:nil];
            }
            break;            
        }
        case NSStreamEventErrorOccurred:{          // can't connect

            NSError *error = [aStream streamError];
            if ([error code]) {
                TTLOGF_WARN(@"NSStreamEventErrorOccurred, <%d:%@>", [[aStream streamError] code], [error localizedDescription]);
//                [self closeStreams];
                [self disconnect];
                [_currentDelegate callBackHandle:[error code] info:[NSString stringWithFormat:@"%@", [error localizedDescription]]]; 
                
            }
            break ;
        }     
        case NSStreamEventNone:
        default:
            TTLOG_WARN(@"NSStreamEventNone, we will nerver get here!!!!");
            assert(NO);
            break;        
    }
}

- (void)dumpToBuffer:(NSOutputStream *)outStream
{
    
    if (_wbHead == _wbTail) {
        // get the data
        NSData *data = (NSData *)[outStream propertyForKey:NSStreamDataWrittenToMemoryStreamKey];
        NSInteger dataLen = [data length];
        
        // TODO this should be processed in the delegate(customer)
        _wbHead = _wbTail = 0;
        memcpy(&_writeBuffer[_wbTail], &dataLen, _packetHeaderSize);
        _wbTail += _packetHeaderSize;
        
        memcpy(&_writeBuffer[_wbTail], [data bytes], dataLen);
        _wbTail += dataLen;
        
        UInt8 type = _writeBuffer[_wbTail - dataLen];
        TTLOGF_INFO(@"Sending message, <type:%d, lenght:%d>", type, dataLen);
    } else {
        TTLOGF_ERROR(@"Data dump to buffer error!! <_wbHead:%d != _wbTail:%d>", _wbHead, _wbTail);
        [self disconnect];
        
        // drop the data
        _wbHead = _wbTail = 0;
        [_currentDelegate callBackHandle:IvanNSEventErrorServerRefuse info:nil];
    }
    
    [outStream close];
}

- (void)writeFlush:(NSOutputStream *)outStream forMode:(NSInteger)forMode
{
    // asynchronous. work in another thread
    
    if (MODE_ASYNC == forMode) {
        // in the thread runloop
        [outStream scheduleInRunLoop:_runLoopOfAsyncListener forMode:NSDefaultRunLoopMode];
    } else {
        // in the main thread runloop
        [self dumpToBuffer:(NSOutputStream *)outStream];
        NSAutoreleasePool *pool = [[NSAutoreleasePool alloc] init];
        [_outStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
        [pool release];
    }
}

- (NSOutputStream *)getWriteFlushOutStream
{
    NSOutputStream * outStream = [NSOutputStream outputStreamToMemory];
    [outStream setDelegate:self];
    [outStream open];
    return outStream;
}

#pragma mark -
#pragma mark maybe deprecated

- (void)threadMain:(id)arg
{
    NSOutputStream *outStream = (NSOutputStream *)arg;
    [outStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
    //[[NSRunLoop currentRunLoop] run];
    
    
    NSLog(@"threadMain");
    return;
    [self dumpToBuffer:(NSOutputStream *)outStream];
    //    
    //    [_outStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSDefaultRunLoopMode];
    //    CFRunLoopRun();
    //    
    if (_wbHead == _wbTail) {
        //[_outStream close];
        return;
    }
    
    //sleep(4);
    //assert(aStream == _outStream);          // in stream should not get here
    while (![_outStream hasSpaceAvailable]){
        //return;
        TTLOG_WARN(@"Outstream not ready");
    }
    
    NSInteger dataLen       = *((NSInteger *)&_writeBuffer[_wbHead]);
    NSInteger packetLen     = dataLen + _packetHeaderSize;
    
    TTDUMP(&_writeBuffer[_wbHead], packetLen);
    NSInteger outLen        = 0;
    NSData *outData         = [_currentDelegate preProcessData:&_writeBuffer[_wbHead] length:packetLen outLength:&outLen];
    
    [StreamPeer writeBytesWithOutStream:_outStream ptr:(UInt8 *)[outData bytes] maxLength:outLen];
    
    _wbHead += packetLen;
    
    _wbTail %= BUFFER_SIZE;
    _wbHead %= BUFFER_SIZE;
    
    if (_wbHead <= _wbTail) {
        // send OK
    } else { 
        TTLOG_ERROR(@"_wbHead > _wbTail, something error occur!");
    } 
    
    //    
    //    [outStream close];
    //    NSRunLoop *runLoop = [NSRunLoop currentRunLoop];
    //    [outStream scheduleInRunLoop:runLoop forMode:NSDefaultRunLoopMode];
    //    CFRunLoopRun();
    
}
/* prepare the integer type data in the buffer, waiting to flush */
- (void)preWriteBytes:(UInt8 *)bytes length:(NSInteger)length
{
    [self writeToBuffer:bytes maxLength:length];
}

- (void)preWriteInt8:(UInt8)int8Value
{
    UInt8 *ptr = (UInt8 *)&int8Value;
    [self writeToBuffer:ptr maxLength:sizeof(UInt8)];
}

- (void)preWriteBoolean:(Boolean)boolValue
{
    UInt8 *ptr = (UInt8 *)&boolValue;
    [self writeToBuffer:ptr maxLength:1];
}

- (void)preWriteInt32:(NSInteger)intValue
{
    UInt8 *ptr = (UInt8 *)&intValue;
    [self writeToBuffer:ptr maxLength:sizeof(NSInteger)];
}

- (void)preWriteInt64:(UInt64)int64Value
{
    UInt8 *ptr = (UInt8 *)&int64Value;
    [self writeToBuffer:ptr maxLength:sizeof(UInt64)];
}

- (void)preWriteString:(NSString *)stringValue
{
    // WARNING the maxLength must pass strlen([stringValue UTF8String], do not pass stringValue.length.
    // Cause Chinese UTF8 take 3 bytes, but take 1 length for NSString length. 
    // When I pass the “你好” strlen([stringValue UTF8String] ＝ 6(or other value), but stringValue.length = 2
    int maxLength = (int)strlen([stringValue UTF8String]);
    [self preWriteInt32:maxLength];
    
    [self writeToBuffer:(UInt8 *)[stringValue UTF8String] maxLength:maxLength];
}

- (void)preWriteArray:(NSArray *)arrayValue
{
    [self preWriteInt32:arrayValue.count];
    for (NSString *stringValue in arrayValue) {
        [self preWriteString:stringValue];
    }
}

/* flush the buffer to the server */
- (void)writeFlush
{
    if (!_hasEstablished) {
        TTLOG_ERROR(@"stream not open yet");
        return;
    }
    
    if (_wbHead == _wbTail)
        return;
    
    //    NSInteger   dataLength = _wbTail > _wbHead ? _wbTail - _wbHead : BUFFER_SIZE - _wbHead;
    NSInteger dataLength      = _wbTail - _wbHead;  
    // TTDUMP(&_writeBuffer[_wbHead], dataLength);
    NSInteger outLength       = 0;
    NSData *outData           = [_currentDelegate preProcessData:&_writeBuffer[_wbHead] length:dataLength outLength:&outLength];
    const UInt8 *outDataUChar = [outData bytes];
    
    NSInteger sendbytes = 0;
    NSInteger gzipHead = 0;
    NSInteger readySendLen = outLength;//[outData length];
    // TTDUMP((UInt8 *)outDataUChar, outLength);
    while ([_outStream hasSpaceAvailable]) {
        if (readySendLen > 0) {
            //[IvanNetStream dumpdata:(gzipBuf + gzipHead) len:readySendLen];
            sendbytes = [_outStream write:(outDataUChar + gzipHead) maxLength:readySendLen];
            //sendbytes = [_outStream write:&_writeBuffer[_wbHead] maxLength:dataLength];
            if (-1 == sendbytes || 0 == sendbytes) {
                // -TODO ERROR HANDLE
                CFStreamError error = CFWriteStreamGetError((CFWriteStreamRef)_outStream);
                [StreamPeer printStreamError:error];
            } else {
                readySendLen -= sendbytes;
                gzipHead += sendbytes;
            }
        } else {
            break;
        }
    }
    
    _wbHead += dataLength;
    
    _wbTail %= BUFFER_SIZE;
    _wbHead %= BUFFER_SIZE;
    
    if (_wbHead == _wbTail) {
        // send OK
        [_outStream close];
    } else { 
        TTLOG_ERROR(@"_wbHead != _wbTail, something error occur!");
    } 
}


- (void)writeToBuffer:(UInt8 *)buffer maxLength:(NSInteger)length
{    
    //NSInteger remainSpace = _wbTail >= _wbHead ? BUFFER_SIZE - (_wbTail - _wbHead) - 1 
    //                                          : _wbHead - _wbTail - 1;
    
    
    if (_wbHead == _wbTail) {
        // no data anymore
        // add the packet header
        _wbHead = _wbTail = 0;
        bzero(&_writeBuffer[_wbHead], _packetHeaderSize);
        _wbTail += _packetHeaderSize;
    } else if (_wbHead > _wbTail) {
        // something is wrong
        TTLOG_ERROR(@"write buffer exception");
    }
    
    NSInteger remainSpace = BUFFER_SIZE - _wbTail;  
    
    if (remainSpace < length) {
        // the buffer is full, flush to the server
        [self writeFlush];
        
        // has space yet
        [self writeToBuffer:buffer maxLength:length];
    }
    
    memcpy(&_writeBuffer[_wbTail], buffer, length);
    _wbTail += length;
}


@end
